package dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.beans.UserJumpStats;
import com.zhang.gmall.utils.ClickHouseUtil;
import com.zhang.gmall.utils.DateTimeUtil;
import com.zhang.gmall.utils.KafkaUtil;
import com.zhang.gmall.utils.MyEnv;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * @title: uj统计分析（用户跳出行为计分析）
 * @author: zhang
 * @date: 2022/3/27 16:28
 */
public class UserJumpVisitDim10sApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取环境
        StreamExecutionEnvironment env = MyEnv.getStreamingEnv(4, false);

        //TODO 2.读取kafka数据，转换为json
        String sourceTopic = "dwd_page_log";
        String groupId = "user_jump_visit_dim_10s_app1";
        KeyedStream<JSONObject, String> kafkaDS = env
                .addSource(KafkaUtil.getKafkaSource(sourceTopic, groupId))
                .map(JSON::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(13))
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("ts");
                                    }
                                })
                )
                .keyBy(json -> json.getJSONObject("common").getString("mid"));

        //TODO 3.定义cep模式序列
        //主要筛选逻辑是匹配跳转数据，让跳出数据进入超时侧输出流，最后提取超时数据
        //跳转行为：第一次进入页面一定时间内访问了其他页面。
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("first")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        //匹配第一次进入页面数据
                        String lastPage = value.getJSONObject("page").getString("last_page_id");
                        return lastPage == null || lastPage.length() <= 0;
                    }
                })
                .next("second")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        //因为上有是页面日志数据所有一定是有pageID的，所以这里可以不用在做过滤。
                        return true;
                    }
                })
                .within(Time.seconds(10L));

        //TODO 4.将序列作用的目标流上
        PatternStream<JSONObject> patternStream = CEP.pattern(kafkaDS, pattern);

        //TODO 5.提取超时数据
        OutputTag<JSONObject> userJumpJsonTag = new OutputTag<JSONObject>("uj") {
        };
        SingleOutputStreamOperator<JSONObject> ujStatsDS = patternStream
                .flatSelect(userJumpJsonTag,
                        new PatternFlatTimeoutFunction<JSONObject, JSONObject>() {
                            @Override
                            public void timeout(Map<String, List<JSONObject>> pattern, long timeoutTimestamp, Collector<JSONObject> out) throws Exception {
                                //提取uj数据  提取超时事件，跳出事件
                                List<JSONObject> list = pattern.get("first");
                                for (JSONObject jsonObject : list) {
                                    //⚠️：虽然调用的是 collector.collect（）方法，但是其实是将超时数据放到参数一 指定的侧输出流
                                    out.collect(jsonObject);
                                }
                            }
                        },
                        new PatternFlatSelectFunction<JSONObject, JSONObject>() {
                            @Override
                            public void flatSelect(Map<String, List<JSONObject>> pattern, Collector<JSONObject> out) throws Exception {
                                //匹配数据不是跳出数据不做提取
                            }
                        });
        //TODO 6.转换为POJO类，分组开窗聚合。
        SingleOutputStreamOperator<UserJumpStats> resultDS = ujStatsDS.getSideOutput(userJumpJsonTag)
                .map(new MapFunction<JSONObject, UserJumpStats>() {
                    @Override
                    public UserJumpStats map(JSONObject logJson) throws Exception {
                        JSONObject common = logJson.getJSONObject("common");
                        return new UserJumpStats(
                                "", "",
                                common.getString("vc"),
                                common.getString("ch"),
                                common.getString("ar"),
                                common.getString("is_new"),
                                1L,
                                logJson.getLong("ts")
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<UserJumpStats>forBoundedOutOfOrderness(Duration.ofSeconds(13))
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserJumpStats>() {
                                    @Override
                                    public long extractTimestamp(UserJumpStats element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                )
                .keyBy(new KeySelector<UserJumpStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(UserJumpStats value) throws Exception {
                        return Tuple4.of(
                                value.getVc(),
                                value.getCh(),
                                value.getAr(),
                                value.getIs_new()
                        );
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<UserJumpStats>() {
                            @Override
                            public UserJumpStats reduce(UserJumpStats value1, UserJumpStats value2) throws Exception {
                                value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                                return value1;
                            }
                        },
                        new WindowFunction<UserJumpStats, UserJumpStats, Tuple4<String, String, String, String>, TimeWindow>() {
                            @Override
                            public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<UserJumpStats> input, Collector<UserJumpStats> out) throws Exception {
                                UserJumpStats userJumpStats = input.iterator().next();
                                //补充窗口时间信息
                                userJumpStats.setStt(DateTimeUtil.toYMDhms(new Date(window.getStart())));
                                userJumpStats.setEdt(DateTimeUtil.toYMDhms(new Date(window.getEnd())));
                                //补充处理时间信息
                                userJumpStats.setTs(System.currentTimeMillis());
                                //向下传递数据
                                out.collect(userJumpStats);
                            }
                        });
        //TODO 7.将数据写入clickhouse
        ujStatsDS.getSideOutput(userJumpJsonTag).print("uj>>>>>");
        resultDS.print("result>>>>>");
        //TODO 8.执行任务
        env.execute("UserJumpVisitDim10sApp");
    }
}
